# RDD 操作：
# 转换（Transformation）： 基于现有的数据集创建一个新的数据集
# ---filter(func)：筛选出满足函数func的元素，并返回一个新的数据集
# ---map(func)：将每个元素传递到函数func中，并将结果返回为一个新的数据集
# ---flatMap(func)：与map()相似，但每个输入元素都可以映射到0或多个输出结果
# ---groupByKey()：应用于(K,V)键值对的数据集时，返回一个新的(K, Iterable)形式的数据集

# 行动（Action）：在数据集上进行运算，返回计算值。
# ---count() 返回数据集中的元素个数
# ---collect() 以数组的形式返回数据集中的所有元素
# ---first() 返回数据集中的第一个元素
# ---take(n) 以数组的形式返回数据集中的前n个元素
# ---reduce(func) 通过函数func（输入两个参数并返回一个值）聚合数据集中的元素
# ---foreach(func) 将数据集中的每个元素传递到函数func中运行*

from pyspark import SparkContext
sc = SparkContext( 'local', 'test')
dir = "D:\Develop\Spark\spark-2.3.2-bin-hadoop2.7"
logFile = "D:\Develop\Spark\spark-2.3.2-bin-hadoop2.7/README.md"

# ---Example 1---
textData = sc.textFile(logFile, 2).cache()
# Transformation: filter
# Action: Count
numAs = textData.filter(lambda line: 'a' in line).count()
numBs = textData.filter(lambda line: 'b' in line).count()
print('Lines with a: %s, Lines with b: %s' % (numAs, numBs))

# ---Example 2---
# 文件系统中加载数据创建RDD
# textFile 就是一个RDD (Resilient Distributed Dataset)
textFile = sc.textFile(dir + "/README.md")
words = textFile.flatMap(lambda line: line.split(" "))
wordCount = words.map(lambda word: (word,1)).reduceByKey(lambda a, b : a + b)
wordCount.foreach(print)

# ---Example 3---